package net.opentsdb.core;

import net.opentsdb.uid.NoSuchUniqueName;
import net.opentsdb.uid.UniqueId;
import net.opentsdb.utils.Assert;
import net.opentsdb.utils.TsdbPair;
import org.hbase.async.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * Tsuid 与 metric、tagkv转换
 *
 */
public final class TsuidExchange {

    private static final Logger LOG = LoggerFactory.getLogger(TsuidExchange.class);

    /**
     * @see #getMetricFromTSUID(TSDB, byte[]) #getTagPairsFromTSUID(TSDB, byte[])
     * */
    public static TsdbPair<String, Map<String,String>> getMetricAndTagFromTSUID(
            final TSDB tsdb, final byte[] tsuid){

        TsdbPair<String, Map<String,String>> tsdbPair = new TsdbPair<>();

        tsdbPair.setKey(getMetricFromTSUID(tsdb, tsuid));
        tsdbPair.setValue(getTagPairsFromTSUID(tsdb, tsuid));

        return tsdbPair;
    }

    /**
     * 根据tsuid获取metric
     *
     * @param tsdb
     * @param tsuid
     * @return
     */
    public static String getMetricFromTSUID(final TSDB tsdb, final byte[] tsuid){
        Assert.notNull(tsuid," tsuid is null ");

        final byte[] metric_uid = Arrays.copyOfRange(tsuid, 0, TSDB.metrics_width());
        String metric;
        try{
            metric = tsdb.getUidName(UniqueId.UniqueIdType.METRIC, metric_uid)
                    .joinUninterruptibly();
        }catch (Exception e){
            LOG.error("get tsuid {} tags failed", tsuid);

            metric = "";
        }

        return metric;
    }

    /**
     * 根据tsuid获取tags
     *
     * @param tsdb
     * @param tsuid
     * @return
     */
    public static Map<String, String> getTagPairsFromTSUID(final TSDB tsdb, final byte[] tsuid) {
        Assert.notNull(tsuid," tsuid is null ");
        Assert.operation(tsuid.length > TSDB.metrics_width(),
                "TSUID is too short, may be missing tags");

        final List<byte[]> tags = new ArrayList<>();
        final int pair_width = TSDB.tagk_width() + TSDB.tagv_width();

        for (int i = TSDB.metrics_width(); i < tsuid.length; i+= pair_width) {
            Assert.operation((i + pair_width) <= tsuid.length,
                    "The TSUID appears to be malformed, improper tag width");
            tags.add(Arrays.copyOfRange(tsuid, i, i + pair_width));
        }

        Map<String, String> map;

        try{
            map = Tags.resolveIdsAsync(tsdb, tags).joinUninterruptibly();
        }catch (Exception e){
            LOG.error("get tsuid {} tags failed", tsuid);
            map = Collections.EMPTY_MAP;
        }

        return map;
    }

    /**
     * metric和tags生成一个tsuid
     *
     * @param tsdb
     * @param metric
     * @param tags
     * @return
     */
    public static byte[] getTsuidFromData(TSDB tsdb, String metric, Map<String, String> tags){
        final byte[] row = rowKeyTemplate(tsdb, metric, tags);

        final byte[] tsuid = UniqueId.getTSUIDFromKey(row, tsdb.metrics_width(),
                Const.TIMESTAMP_BYTES);

        return tsuid;
    }

    private static byte[] rowKeyTemplate(final TSDB tsdb, final String metric,
                                 final Map<String, String> tags) {
        final short metric_width = tsdb.metrics_width();
        final short tag_name_width = tsdb.tagk_width();
        final short tag_value_width = tsdb.tagv_width();
        final short num_tags = (short) tags.size();

        int row_size = (Const.SALT_WIDTH() + metric_width + Const.TIMESTAMP_BYTES
                + tag_name_width * num_tags + tag_value_width * num_tags);
        final byte[] row = new byte[row_size];

        short pos = (short) Const.SALT_WIDTH();

        copyInRowKey(row, pos,
                (tsdb.getConfig().auto_metric() ? tsdb.metrics.getOrCreateId(metric)
                        : tsdb.metrics.getId(metric)));
        pos += metric_width;

        pos += Const.TIMESTAMP_BYTES;

        for (final byte[] tag : resolveAllInternal(tsdb, tags, true)) {
            copyInRowKey(row, pos, tag);
            pos += tag.length;
        }
        return row;
    }

    private static void copyInRowKey(final byte[] row, final short offset,
                                     final byte[] bytes) {
        System.arraycopy(bytes, 0, row, offset, bytes.length);
    }


    private static ArrayList<byte[]> resolveAllInternal(final TSDB tsdb,
                                                final Map<String, String> tags,
                                                final boolean create)
            throws NoSuchUniqueName {
        final ArrayList<byte[]> tag_ids = new ArrayList<byte[]>(tags.size());
        for (final Map.Entry<String, String> entry : tags.entrySet()) {
            final byte[] tag_id = (create && tsdb.getConfig().auto_tagk()
                    ? tsdb.tag_names.getOrCreateId(entry.getKey())
                    : tsdb.tag_names.getId(entry.getKey()));
            final byte[] value_id = (create && tsdb.getConfig().auto_tagv()
                    ? tsdb.tag_values.getOrCreateId(entry.getValue())
                    : tsdb.tag_values.getId(entry.getValue()));
            final byte[] thistag = new byte[tag_id.length + value_id.length];
            System.arraycopy(tag_id, 0, thistag, 0, tag_id.length);
            System.arraycopy(value_id, 0, thistag, tag_id.length, value_id.length);
            tag_ids.add(thistag);
        }
        // Now sort the tags.
        Collections.sort(tag_ids, Bytes.MEMCMP);
        return tag_ids;
    }
}
